package nsqframe

import (
	"errors"
	"net"
	"time"

	"github.com/nsqio/go-nsq"

	"github.com/xlvector/dlog"
)

type NFProducer struct {
	productCh   chan []byte
	conf        *NFPConfig
	nsqProducer *nsq.Producer

	send      int64 //发送nsq个数
	sendError int64 //发送失败个数
}

func newNsqProducer(nsqserver string) *nsq.Producer {
	config := nsq.NewConfig()
	config.LocalAddr, _ = net.ResolveTCPAddr("tcp", ":0")
	p, err := nsq.NewProducer(nsqserver, config)
	if err != nil {
		dlog.Fatal("new nsq producer fail! %v", err)
	}
	return p
}

func NewNFProducer(conf *NFPConfig) *NFProducer {
	p := &NFProducer{
		productCh:   make(chan []byte, conf.CacheLen),
		conf:        conf,
		nsqProducer: newNsqProducer(conf.Nsq.Server),
	}

	return p
}

func (p *NFProducer) work(idx int) {
	dlog.Info("work: %d", idx)
	var product []byte
	var ok bool = true
	var msgcache = make([][]byte, 0, PRODUCER_CACHE_LEN)
	var msgcacheLen int64 = 0

	tc := time.NewTicker(time.Second)
	for {
		select {
		case product, ok = <-p.productCh:
			if !ok {
				goto lab_publish
			}
			//dlog.Info("msg: %v", product)
			msgcache = append(msgcache, product)
			if len(msgcache) < PRODUCER_CACHE_LEN {
				continue
			}
		case <-tc.C:
		}

	lab_publish:
		msgcacheLen = int64(len(msgcache))
		if msgcacheLen > 0 {
			err := p.nsqProducer.MultiPublish(p.conf.Nsq.Topic, msgcache)
			if err != nil {
				dlog.Error("multi publish fail! %v -- %v", err, msgcache)
				p.sendError += msgcacheLen
			} else {
				p.send += msgcacheLen
			}
			msgcache = msgcache[0:0]
		}

		if !ok {
			dlog.Info("recive close product! %d", idx)
			return
		}
	}
}

func (p *NFProducer) Start() {
	for i := 0; i < p.conf.Concurrent; i++ {
		go p.work(i)
	}
}

func (p *NFProducer) Stop() {
	close(p.productCh)
	p.nsqProducer.Stop()
}

func (p *NFProducer) Product(data []byte) error {
	select {
	case p.productCh <- data:
		return nil
	default:
		return errors.New("cache full!")
	}
}

/*
   return val:  返回发送失败数据
*/
func (p *NFProducer) ProductMany(data [][]byte) ([][]byte, error) {
	for i, d := range data {
		select {
		case p.productCh <- d:
		default:
			return data[i:], errors.New("cache full!")
		}
	}
	return nil, nil
}
